package com.distribute.core.init;

import com.distribute.core.netty.LocalCacheClientInitializer;
import com.distribute.core.netty.listener.ConnectSuccessListener;
import com.distribute.core.netty.retry.ExponentialBackOffRetry;
import com.distribute.core.netty.retry.RetryPolicy;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;

/**
 * 缓存上报客户端
 *
 * @author chennengcheng
 * @date 2021-09-13 17:37
 * @remark
 */
public class CacheReportClient {

    private final Bootstrap bootstrap = new Bootstrap();

    private final EventLoopGroup workerGroup =  new NioEventLoopGroup();

    public static ChannelHandlerContext cacheClientHandler;

    private String host;

    private Integer port;

    private Channel channel;

    /** 重连策略 */
    private RetryPolicy retryPolicy;

    private CacheReportClient(String host, Integer port, RetryPolicy retryPolicy) {
        this.retryPolicy = retryPolicy;
        this.host = host;
        this.port = port;
    }

    public CacheReportClient(String host, Integer port) {
        this(host, port, new ExponentialBackOffRetry(RetryPolicy.BASE_RECONNECT_TIME, Integer.MAX_VALUE, 60 * 1000));
    }

    /**
     * 初始化 nettyServer
     */
    public void initNettyServer(){
        try {
            // 客户端启动类程序
            bootstrap.group(workerGroup);
            bootstrap.channel(NioSocketChannel.class);
            // 参数：Socket的标准参数（key，value），可自行百度 保持呼吸，不要断气！ 但时间较长，要保证灵敏还是需要 自己实现心跳
            bootstrap.option(ChannelOption.SO_KEEPALIVE, true);
            // 自定义客户端Handle（客户端在这里搞事情）
            bootstrap.handler(new LocalCacheClientInitializer(CacheReportClient.this));
            // 开启客户端监听，连接到远程节点，阻塞等待直到连接完成
            this.connect();
            // 阻塞等待数据，直到channel关闭(客户端关闭)
            this.channel.closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 向远程TCP服务器请求连接
     */
    public void connect() {
        synchronized (bootstrap) {
            ChannelFuture future = bootstrap.connect(host, port);
            future.addListener(getConnectionListener());
            // 添加启动成功后监听器
            future.addListener(new ConnectSuccessListener());
            this.channel = future.channel();
        }
    }

    public RetryPolicy getRetryPolicy() {
        return retryPolicy;
    }


    private ChannelFutureListener getConnectionListener() {
        return future -> {
            if (!future.isSuccess()) {
                this.channel.pipeline().fireChannelInactive();
            }
        };
    }
}
